package com.wsjj.yjh.watermark;

import com.wsjj.yjh.WaterSensor;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class CustomPeriodicWatermarkExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        DataStreamSource<String> source = env.socketTextStream("hadoop102", 7777);
        SingleOutputStreamOperator<WaterSensor> map = source.map(value -> {
            String[] split = value.split(",");
            WaterSensor waterSensor = new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
            return waterSensor;
        });


        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator
                = map.assignTimestampsAndWatermarks(new CustomWatermarkStrategy());
        waterSensorSingleOutputStreamOperator.print();
        env.execute();
    }

    private static class CustomWatermarkStrategy implements WatermarkStrategy<WaterSensor> {

        @Override
        public WatermarkGenerator<WaterSensor> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
            return new WatermarkGenerator<WaterSensor>() {
                private Long delayTime = 3000L;
                private Long maxTs = -Long.MAX_VALUE+delayTime+1L;


                @Override
                public void onEvent(WaterSensor event, long eventTimestamp, WatermarkOutput output) {
                    System.out.println("eventTimestamp:"+eventTimestamp);
                    long maxTs = Math.max(event.getTs(), eventTimestamp);
                    System.out.println(maxTs);

                }

                @Override
                public void onPeriodicEmit(WatermarkOutput output) {
                    System.out.println("水位线时间：" + (maxTs - delayTime - 1L));
                    output.emitWatermark(new Watermark(maxTs - delayTime - 1L));

                }
            };
        }

        @Override
        public TimestampAssigner<WaterSensor> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
            return new SerializableTimestampAssigner<WaterSensor>() {
                @Override
                public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                    return element.getTs();
                }
            };
        }
    }
}
